package com.atguigu0.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @description: sparkStreaming案例,无状态,每次统计单词的个数不会进行合并,5s内数据互不相干.
 * @time: 2020/6/15 20:54
 * @author: baojinlong
 **/
object WordCountStreaming {
  def main(args: Array[String]): Unit = {
    // 创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")
    // 创建SteamingContext
    val ssc = new StreamingContext(sparkConf, Seconds(3))
    // 创建Dstream
    val line: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)
    // 压平数据
    val word: DStream[String] = line.flatMap(_.split(" "))
    // 将单词转成元组
    val wordAndOne: DStream[(String, Int)] = word.map((_, 1))
    // 按照key聚合value
    val wordAndCount: DStream[(String, Int)] = wordAndOne.reduceByKey(_ + _)
    // 打印数据
    wordAndCount.print
    // 开启sparkStreaming
    ssc.start()
    // 等待执行
    ssc.awaitTermination()
  }
}
